package cn.iocoder.yudao.module.bpm.message.rocketmq.custom.processTask;


import cn.iocoder.yudao.framework.mq.rocketmq.api.consumer.handler.EnhanceMessageHandler;
import cn.iocoder.yudao.module.bpm.message.handle.ProcessTaskMessageHandle;
import cn.iocoder.yudao.module.bpm.message.rocketmq.dto.RocketMqBpmProcessTaskCancelledMessage;
import cn.iocoder.yudao.module.bpm.message.consts.ProcessTaskStreamKey;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;

@Slf4j
@RequiredArgsConstructor
@RocketMQMessageListener(topic = ProcessTaskStreamKey.PROCESS_INSTANCE_TASK_CLEAN_MSG_STREAM,
        consumerGroup = ProcessTaskStreamKey.PROCESS_INSTANCE_TASK_CLEAN_MSG_GROUP +"_" + "${spring.application.name}",
        messageModel = MessageModel.CLUSTERING //消费模式 默认CLUSTERING集群  BROADCASTING:广播（接收所有信息）
)
public class RocketMqProcessTaskCancelledCustom extends EnhanceMessageHandler<RocketMqBpmProcessTaskCancelledMessage> implements RocketMQListener<RocketMqBpmProcessTaskCancelledMessage> {


    private final ProcessTaskMessageHandle processTaskMessageHandle;

    /**
     * 消息处理
     *
     * @param message 待处理消息
     * @throws Exception 消费异常
     */
    @Override
    protected void handleMessage(RocketMqBpmProcessTaskCancelledMessage message)  {
        try {
            log.info("消费消息：{}", message);
            processTaskMessageHandle.onProcessInstanceTaskCleanMsg(message.getBpmProcessMessage());
        } catch (Exception e) {
            log.error("MQ消费失败",e);
            throw new RuntimeException(e.getMessage());
        }
    }

    @Override
    public void onMessage(RocketMqBpmProcessTaskCancelledMessage message) {
        try {
            dispatchMessage(message);
        } catch (Exception e) {
            log.error("MQ消费失败", e);
            throw new RuntimeException(e.getMessage());
        }
    }

}
